package com.mjf.day2

import org.apache.flink.api.common.functions.FilterFunction
import org.apache.flink.streaming.api.scala._

object FilterExample {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    val stream: DataStream[SensorReading] = env.addSource(new SensorSource)

    // 方式一
    stream.filter(_.id.equals("sensor_1")).print()

    // 方式二
    stream.filter(new MyFilterFunction()).print()

    // 方式三
    stream.filter(new FilterFunction[SensorReading] {
      override def filter(t: SensorReading): Boolean = t.id.equals("sensor_1")
    }).print()

    env.execute("FilterExample")
  }

  // filter算子的输入和输出类型是一样的
  class MyFilterFunction extends FilterFunction[SensorReading] {
    override def filter(t: SensorReading): Boolean = t.id.equals("sensor_1")
  }
}
